package com.intct.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-11-13 08:33
 */
public class MysqlLookupHBase {
    public static void main(String[] args) {
        // 1. 初始化流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L, org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE);

        // 2. 初始化表执行环境（适配 Flink 1.15+）
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 注册自定义函数
        tableEnv.createTemporarySystemFunction("SUM_FUNC", SumFunction.class);

        // mysql-cdc
        String createCdcSource = "CREATE TABLE orders (" +
                "  id INT," +
                "  order_no STRING," +
                "  proc_time AS PROCTIME()," +
                "  PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (" +
                "  'connector' = 'mysql-cdc'," +
                "  'hostname' = 'cdh-node'," +
                "  'port' = '13306'," +
                "  'username' = 'root'," +
                "  'password' = 'Test_090110'," +
                "  'database-name' = 'travel'," +
                "  'table-name' = 'order_info'," +
                "  'scan.startup.mode' = 'initial'" +
                ")";
        tableEnv.executeSql(createCdcSource);

        // hbase
        String createHBaseSource = "CREATE TABLE order_state (" +
                " rowkey STRING," +
                " info ROW<name STRING, state STRING>," +
                " PRIMARY KEY (rowkey) NOT ENFORCED" +
                ") WITH (" +
                " 'connector' = 'hbase-2.2'," +
                " 'table-name' = 'dim:dim_order_state'," +
                " 'zookeeper.quorum' = 'cdh-node:2181'" +
                ")";
        tableEnv.executeSql(createHBaseSource);


        tableEnv.executeSql("SELECT" +
                "    o.id" +
                "    , SUM_FUNC(o.id, 10) calac" +
                "    , o.order_no" +
                "    , s.info.name" +
                "    , s.info.state" +
                " FROM orders o " +
                " JOIN order_state FOR SYSTEM_TIME AS OF o.proc_time AS s" +
                "   ON o.order_no = s.rowkey").print();
    }
}